/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.analysys.presto.connector.hbase.utils;

import com.analysys.presto.connector.hbase.schedule.ConditionInfo;
import com.facebook.airlift.log.Logger;
import com.facebook.presto.spi.type.ArrayType;
import com.facebook.presto.spi.type.BigintType;
import com.facebook.presto.spi.type.BooleanType;
import com.facebook.presto.spi.type.DecimalType;
import com.facebook.presto.spi.type.DoubleType;
import com.facebook.presto.spi.type.IntegerType;
import com.facebook.presto.spi.type.TimestampType;
import com.facebook.presto.spi.type.Type;
import com.facebook.presto.spi.type.VarcharType;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos;
import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
import org.apache.hadoop.hbase.snapshot.SnapshotManifest;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;

import static com.analysys.presto.connector.hbase.utils.Constant.ARRAY_STRING_SPLITTER;
import static com.analysys.presto.connector.hbase.utils.Constant.ConditionOper;
import static com.analysys.presto.connector.hbase.utils.Constant.DECIMAL_DEFAULT_PRECISION;
import static com.analysys.presto.connector.hbase.utils.Constant.DECIMAL_DEFAULT_SCALE;

/**
 * utils
 *
 * @author wupeng
 * @date 2019/01/29
 */

public class Utils
{
    public static final Logger logger = Logger.get(Utils.class);

    /**
     * Add zero prefix to a salt string
     *
     * @param key       key
     * @param missCount missCount
     * @return standard salt value
     */

    public static String addZeroPrefix(String key, int missCount)
    {
        for (int j = missCount; j > 0; j--) {
            key = "0" + key;
        }
        return key;
    }

    /**
     * Find the presto type of column you configured in json file by type flag.
     *
     * @param type The type value that configured in json file.
     * @return type in presto
     */

    public static Type matchType(String type)
    {
        if (type == null) {
            return VarcharType.VARCHAR;
        }
        switch (type.toLowerCase(Locale.ENGLISH)) {
            /*case "string":
                return VarcharType.VARCHAR;*/
            case "int":
                return IntegerType.INTEGER;
            case "integer":
                return IntegerType.INTEGER;
            case "bigint":
                return BigintType.BIGINT;
            case "double":
                return DoubleType.DOUBLE;
            case "boolean":
                return BooleanType.BOOLEAN;
            case "array<string>":
                return new ArrayType(VarcharType.VARCHAR);
            case "timestamp":
                // return TimestampType.TIMESTAMP;
            case "datetime":
                return TimestampType.TIMESTAMP;
            case "number":
                return DecimalType.createDecimalType(DECIMAL_DEFAULT_PRECISION, DECIMAL_DEFAULT_SCALE);
            default:
                return VarcharType.VARCHAR;
        }
    }

    /**
     * Whether sql constraint contains conditions like "rowKey='xxx'" or "rowKey in ('xxx','xxx')"
     *
     * @return true if this sql is batch get.
     */

    public static boolean isBatchGet(List<ConditionInfo> conditions, String rowKeyColName)
    {
        if (conditions != null) {
            for (ConditionInfo cond : conditions) {
                if (rowKeyColName.equals(cond.getColName())
                        && cond.getOperator() == ConditionOper.EQ) {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * Copy contents in ${srcAry} from position ${srcPos} for ${length} bytes.
     *
     * @param srcAry source array
     * @param srcPos start position
     * @param length length
     * @return copied byte array
     */

    public static byte[] arrayCopy(byte[] srcAry, int srcPos, int length)
    {
        byte[] destAry = new byte[length];
        System.arraycopy(srcAry, srcPos, destAry, 0, length);
        return destAry;
    }

    public static Configuration getHadoopConf(String zookeeperQuorum, String zookeeperClientPort)
    {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", zookeeperQuorum);
        conf.set("hbase.zookeeper.property.clientPort", zookeeperClientPort);
        conf.set("hbase.cluster.distributed", "true");
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        conf.set("hbase.client.retries.number", "3");
        return conf;
    }

    public static List<RegionInfo> getRegionInfosFromManifest(SnapshotManifest manifest)
    {
        ArrayList<SnapshotProtos.SnapshotRegionManifest> regionManifests = new ArrayList<>(manifest.getRegionManifests());

        ArrayList<RegionInfo> regionInfo = new ArrayList<>(regionManifests.size());

        for (SnapshotProtos.SnapshotRegionManifest regionManifest : regionManifests) {
            RegionInfo hri = ProtobufUtil.toRegionInfo(regionManifest.getRegionInfo());
            if (hri.isOffline() && (hri.isSplit() || hri.isSplitParent())) {
                continue;
            }
            regionInfo.add(hri);
        }
        return regionInfo;
    }

    /**
     * get region infos
     *
     * @param zookeeperQuorum     zookeeper quorum
     * @param zookeeperClientPort zookeeper client port
     * @param hBaseRootDir        HBase root dir
     * @param snapshotName        snapshot name
     * @return region info list
     * @throws IOException IOException
     */

    public static List<RegionInfo> getRegionInfos(String zookeeperQuorum,
            String zookeeperClientPort,
            String hBaseRootDir,
            String snapshotName)
            throws IOException
    {
        try {
            Configuration conf = Utils.getHadoopConf(zookeeperQuorum, zookeeperClientPort);
            Path root = new Path(hBaseRootDir);
            FileSystem fs = FileSystem.get(conf);
            Path snapshotDir = SnapshotDescriptionUtils.getCompletedSnapshotDir(snapshotName, root);
            SnapshotProtos.SnapshotDescription snapshotDesc = SnapshotDescriptionUtils.readSnapshotInfo(fs, snapshotDir);
            SnapshotManifest manifest = SnapshotManifest.open(conf, fs, snapshotDir, snapshotDesc);
            return Utils.getRegionInfosFromManifest(manifest);
        }
        catch (IOException ex) {
            logger.error("get region info error: " + ex.getMessage(), ex);
            throw ex;
        }
    }

    /**
     * 去掉array<string>中\001分隔的元素前的一个空格
     * presto查询出来的值是"aaa\001 bbb\001 ccc"，元素之间会带上一个空格
     *
     * @param value value
     * @return string
     */

    public static String removeExtraSpaceInArrayString(String value)
    {
        StringBuilder buff = new StringBuilder();
        String[] ss = value.split(ARRAY_STRING_SPLITTER);
        for (int j = 0; j < ss.length; j++) {
            String ele = ss[j];
            if (j > 0) {
                ele = ele.substring(1);
            }
            else {
                buff.append(ARRAY_STRING_SPLITTER);
            }
            buff.append(ele);
        }
        return buff.toString();
    }

    public static boolean isEmpty(String str)
    {
        return str == null || "".equals(str);
    }
}
